package com.xahj.bd2104.spark.core

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * Created with IntelliJ IDEA.
 * Author: Amos
 * E-mail: amos@amoscloud.com
 * Date: 2021/9/15
 * Time: 8:33
 * Description: 
 */
object SparkCoreDemo8_actions {
  def main(args: Array[String]): Unit = {
    // 3. 行动算子 action
    //    RDD  =>  !RDD

    // RDD是懒执行
    // RDD的依赖中，如果最终没有调用action算子
    // 那么整个RDD的依赖链条仅仅记录依赖关系 不执行真正的运算
    // sc=>RDD
    // RDD.transformations => RDD1
    // RDD1.transformations => RDD2
    // RDD2.transformations => RDD3
    // RDD3.actions =>

    val sc = new SparkContext(new SparkConf()
      .setAppName("demo8")
      .setMaster("local[2]"))

    sc.setCheckpointDir("C:\\Users\\Amos\\Desktop\\checkpoint")


    val rdd = sc.makeRDD(1 to 10)

    val rdd1 = rdd.map(x => {
      println(Thread.currentThread().getId + "--" + x)
      x + 1
    })

    // 1.  reduce
    // 传入聚合函数  对RDD中的元素进行聚合
    //    reduce不能改变元素类型
    // def reduce(f: (T, T) => T): T =
    val res1: Int = rdd.reduce(_ * _)
    println(res1)


    // 2. collect
    // spark-shell中常用于将RDD的结果做展示
    // def collect(): Array[T] =
    val ints = rdd.collect()

    // 3. count
    //  获取RDD中元素的数量
    val l: Long = rdd.count()

    // 4. first
    //    def first(): T =
    val i = rdd.first()

    // 5. take
    //def take(num: Int): Array[T] =
    val res5 = rdd.take(1)

    // 6. takeSample
    //  def takeSample(
    //      withReplacement: Boolean,
    //      num: Int,
    //      seed: Long = Utils.random.nextLong): Array[T] =
    val res6 = rdd.takeSample(true, 6)

    // 7. takeOrdered
    val rdd2 = sc.makeRDD(List("11", "2", "111", "22", "10000"))
    //    def takeOrdered(num: Int): Array[T] =
    val res7 = rdd2.takeOrdered(1)
    // 类似  rdd2.sortBy(x=>x).take(1)


    // 8. 将RDD存储到文件
    //     rdd2.saveAsTextFile("")
    //     rdd2.saveAsObjectFile("")

    //     rdd2.map(x=>(1,x)).saveAsSequenceFile("")


    // 9. countByKey

    val rdd9: RDD[String] = sc.makeRDD(List("asdasd", "hyuhtgrf", "htesda"))
    val map = rdd9.flatMap(x => x.toCharArray)
      .map((_, 1))
      //      .countByKey()
      .reduceByKey(_ + _)
      .collect()
      .toMap


    // 10. foreach
    //def foreach(f: T => Unit): Unit =
    // foreach用于自定义RDD的元素处理
    //   例如测试时通常输出到控制台
    //   或者将RDD中元素输出到 数据库、自定义文件等
    val unit = rdd.foreach(x => {

    })


  }
}
